package com.fadadianqi.test;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.LogFactory;

import com.fadadianqi.logistics.admin.net.packet.SMSPacket;
import com.fadadianqi.commons.JSONProvider;
import com.yeqiangwei.commons.util.Utils;

public class Connection {

	public static void main(String[] args) {
		
		Connection conn = new Connection();
		try {
			conn.connect();
			for(int i=0; i<100; i++) {
				conn.send(new SMSPacket(JSONProvider.toJSON(i)).toBytes());
				Thread.sleep(3000);
			}
			conn.send(new SMSPacket(JSONProvider.toJSON("##########################################")).toBytes());
			System.out.println("##########################################");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private static final org.apache.commons.logging.Log Log = LogFactory
			.getLog(Connection.class);

	public static final String SERVER_IP = "192.168.1.103";
	public static final int SERVER_PORT = 5222;
	private static ConcurrentHashMap<Long, byte[]> sendBuffer = new ConcurrentHashMap<Long, byte[]>();
	private ByteBuffer readBuffer = ByteBuffer.allocate(8192);
	private Lock lock = new ReentrantLock(false);
	public long latestTime = 0; // 最后活动时间

	private String address;
	private int port;
	private SocketChannel socketChannel;
	private Selector selector;
	private boolean isConnected;

	public Connection() {
		this.address = SERVER_IP;
		this.port = SERVER_PORT;
		this.isConnected = Boolean.FALSE;
	}

	public Connection(String address, Integer port) {
		this.address = address;
		this.port = port;
		this.isConnected = Boolean.FALSE;
	}

	public void connect() throws Exception {
		selector = Selector.open();
		SocketAddress address = new InetSocketAddress(this.address, this.port);
		System.out.println("#connect#, SocketChannel opening...");
		socketChannel = SocketChannel.open(address);
		socketChannel.configureBlocking(Boolean.FALSE);
		socketChannel.register(selector, SelectionKey.OP_CONNECT|SelectionKey.OP_READ|SelectionKey.OP_WRITE);
		System.out.println("#connect#, SocketChannel register selector.");
		while (true) {
			if (socketChannel.finishConnect()) {
				System.out.println("#connect#, finish connect...");
				break;
			}
		}
		latestTime = new Date().getTime();
		isConnected = Boolean.TRUE;
		listen();
		//SessionManager.addSession(new ClientSession(null));
	}

	private void finishConnection(final SelectionKey key) throws IOException {
		System.out.println("#finishConnection#");
		SocketChannel socketChannel = (SocketChannel) key.channel();
		socketChannel.finishConnect();
		key.interestOps(SelectionKey.OP_WRITE);
	}

	private void listen() {
		Runnable r = new Runnable() {
			@Override
			public void run() {
				System.out.println("isConnected:" + isConnected);
				while (isConnected) {
					try {
						Thread.sleep(500);
						System.out.println("selector.isOpen():" + selector.isOpen());
						if (!selector.isOpen()) {
							continue;
						}
						selector.select();
						Iterator<?> it = selector.selectedKeys().iterator();
						while (it.hasNext()) {
							SelectionKey key = (SelectionKey) it.next();
							System.out.println("key.isValid():" + key.isValid()+ " " + key.toString());
							if (!key.isValid()) {
								continue;
							}
							// Check what event is available and deal with it
							System.out.println("key.isConnectable():"
									+ key.isConnectable());
							if (key.isConnectable()) {
								finishConnection(key);
							}
							System.out.println("key.isReadable():" + key.isReadable());
							if (key.isReadable()) {
								read(key);
							}
							System.out.println("key.isWritable():" + key.isWritable());
							if (key.isWritable()) {
								lock.lock();
								write(key);
								lock.unlock();
							}
							it.remove();
						}
					} catch (Exception e) {
						Log.error(e.toString());
						e.printStackTrace();
					}
				}
			}
		};
		new Thread(r).start();
	}

	public boolean isClosed() {
		return !this.isConnected;
	}

	private void write(SelectionKey key) {
		if (!key.isValid()) {
			System.out.println("Connection#write#1, key is invalid.");
			return;
		}
		if (isClosed()) {
			System.out.println("Connection#write#1, isConnected: " + socketChannel.isConnected()
					+ ", isOpen: " + socketChannel.isOpen());
			return;
		}
		System.out.println("sendBuffer.isEmpty():" + sendBuffer.isEmpty());
		while (!sendBuffer.isEmpty()) {
			Iterator<Long> bufKeys = sendBuffer.keySet().iterator();
			while (bufKeys.hasNext()) {
				Long i = bufKeys.next();
				byte[] bytes = sendBuffer.get(i);
				if (bytes == null || bytes.length == 0) {
					continue;
				}
				System.out.println( "#write#, bytes length: " + bytes.length);
				ByteBuffer bb = ByteBuffer.wrap(bytes);
				try {
					int w = socketChannel.write(bb);
					System.out.println("#write#, length: " + bytes.length
							+ ", written: " + w + ", remaining: "
							+ bb.remaining());
					while (bb.remaining() > 0) {
						if (isClosed() || !key.isValid()) {
							System.out.println("#write#, length: isClosed() || !key.isValid()");
							break;
						}
						System.out.println("#write#, length: " + bytes.length
								+ ", written: "
								+ (bytes.length - bb.remaining()));
						socketChannel.write(bb);
					}
				} catch (Exception e) {
					Log.error(e.toString());
				} finally {
					bufKeys.remove();
					sendBuffer.remove(i);
					bb.clear();
				}
			}
		}
		try {
			socketChannel.register(selector, SelectionKey.OP_CONNECT|SelectionKey.OP_READ|SelectionKey.OP_WRITE);
		} catch (ClosedChannelException e) {
			e.printStackTrace();
		}  
	}

	private void read(SelectionKey key) throws Exception {
		if (!key.isValid()) {
			System.out.println("Connection#read#, SelectionKey is invalid");
			return;
		}
		// Clear out our read buffer so it's ready for new data
		this.readBuffer.clear();
		int read = 0;
		int c = 0;
		while ((read = socketChannel.read(readBuffer)) != -1) {
			if (read == 0) {
				break;
			}
			c++;
		}
		readBuffer.flip();
		latestTime = new Date().getTime();
		if (c > 0) {
			// channelHandler.handle(readBuffer);
		}
		socketChannel.register(selector, SelectionKey.OP_CONNECT|SelectionKey.OP_READ|SelectionKey.OP_WRITE);
	}

	public void close() {
		this.isConnected = false;
		if (socketChannel != null) {
			try {
				System.out.println("Connection#close#, this: " + this.toString());
				socketChannel.close();
			} catch (Exception e) {
				System.out.println(e.toString());
			} finally {
				socketChannel = null;
			}
		}
	}

	public void send(byte[] bytes) {
		lock.lock();
		long key = new Date().getTime()+Utils.getRandomNumber(0, Integer.MAX_VALUE);
		System.out.println("send(byte[] bytes, key:" + key);
		sendBuffer.put(key, bytes);
		try {
			Thread.sleep(1);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		lock.unlock();
	}

	public long getLatestTime() {
		return latestTime;
	}
}
